use crate::db::Db;
use crate::models::*;
use crate::redis::*;
use crate::util::*;
use anyhow::Result;
use anyhow::{anyhow, bail};
use diesel::prelude::*;
use dotenv::var;
use redis::JsonAsyncCommands;
use rumqttc::v5::AsyncClient;
use salvo::{Error, oapi::extract::*, prelude::*, routing::PathFilter};
use serde::{Deserialize, Serialize};
use tokio_graceful_shutdown::SubsystemHandle;
use tracing::info;

pub async fn run(subsys: SubsystemHandle, db: Db, redis: Redis, client: AsyncClient) -> Result<()> {
    let acceptor = TcpListener::new(format!("0.0.0.0:{}", var("PORT").unwrap()))
        .bind()
        .await;
    let router_api = Router::with_hoop(BasicAuth::new(Validator))
        .hoop(affix_state::inject(db))
        .hoop(affix_state::inject(redis))
        .hoop(affix_state::inject(client))
        .filter(PathFilter::new("api"))
        .get(health)
        .push(
            Router::with_path("users")
                .push(Router::with_path("lis").post(user_lis))
                .push(Router::with_path("register").post(user_add))
                .push(Router::with_path("upd").post(user_upd))
                .push(Router::with_path("change_pass").post(user_upd_pass))
                .push(Router::with_path("{id}").get(user_get).delete(user_del)),
        )
        .push(
            Router::with_path("spaces")
                .push(Router::with_path("lis").post(space_lis))
                .push(Router::with_path("add").post(space_add))
                .push(Router::with_path("upd").post(space_upd))
                .push(Router::with_path("{id}").get(space_get).delete(space_del))
                .push(Router::with_path("enter/{space_id}").post(space_enter))
                .push(Router::with_path("leave/{space_id}").post(space_leave))
                .push(Router::with_path("state/{space_id}").get(space_state))
                .push(Router::with_path("state/patch/{space_id}").post(space_state_patch)),
        )
        .push(
            Router::with_path("bookings")
                .push(Router::with_path("lis").post(booking_lis))
                .push(Router::with_path("add").post(booking_add))
                .push(Router::with_path("upd").post(booking_upd))
                .push(
                    Router::with_path("{space_id}/{user_id}")
                        .get(booking_get)
                        .delete(booking_del),
                ),
        );
    let crate_name = env!("CARGO_CRATE_NAME");
    let router = Router::with_path(crate_name)
        .push(
            OpenApi::new("api", "0.0.1")
                .merge_router_with_base(&router_api, format!("/{}", crate_name))
                .into_router("api-doc/openapi.json"),
        )
        .push(
            SwaggerUi::new(format!("/{}/api-doc/openapi.json", crate_name))
                .into_router("swagger-ui"),
        )
        .push(router_api)
        .push(
            Router::with_path("view/{**}").get(
                StaticDir::new(["static"])
                    .defaults("index.html")
                    .auto_list(true),
            ),
        );
    info!("ROUTER -- \n{:?}", router);
    let server = Server::new(acceptor);
    let handle = server.handle();
    tokio::spawn(async move {
        subsys.on_shutdown_requested().await;
        handle.stop_graceful(Some(std::time::Duration::from_secs(10)));
    });
    server.serve(Service::new(router).hoop(Logger::new())).await;
    Ok(())
}

struct Validator;

impl BasicAuthValidator for Validator {
    async fn validate(&self, username: &str, _password: &str, depot: &mut Depot) -> bool {
        if username == "teacher" {
            depot.insert("user_id", 1_i64);
        }
        if username == "student" {
            depot.insert("user_id", 2_i64);
        }
        true
    }
}

fn obtain_db(depot: &Depot) -> anyhow::Result<&Db> {
    let v = depot
        .obtain::<Db>()
        .map_err(|_| anyhow!("cannot get 'db' from depot."))?;
    Ok(v)
}

fn obtain_redis(depot: &Depot) -> anyhow::Result<Redis> {
    let v = depot
        .obtain::<Redis>()
        .map_err(|_| anyhow!("cannot get 'redis' from depot."))?;
    Ok(v.clone())
}

fn obtain_mqtt_client(depot: &Depot) -> anyhow::Result<&AsyncClient> {
    let v = depot
        .obtain::<AsyncClient>()
        .map_err(|_| anyhow!("cannot get 'mqtt_client' from depot."))?;
    Ok(v)
}

fn obtain_user_id(depot: &Depot) -> anyhow::Result<i64> {
    let user_id = depot
        .get::<i64>("user_id")
        .map_err(|_| anyhow!("cannot get 'user_id' from depot."))
        .copied()?;
    Ok(user_id)
}

#[endpoint]
pub async fn health() -> &'static str {
    "Hello!"
}

#[endpoint]
pub async fn user_add(to_add: JsonBody<UserToAdd>, depot: &Depot) -> Result<Json<User>, Error> {
    use crate::schema::users;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let rec = diesel::insert_into(users::table)
        .values(Some(to_add.clone()))
        .get_result::<User>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn user_upd(to_upd: FormBody<UserToUpd>, depot: &Depot) -> Result<Json<User>, Error> {
    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let rec = to_upd
        .save_changes::<User>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn user_upd_pass(
    to_upd: FormBody<UserToUpdPass>,
    depot: &Depot,
) -> Result<Json<User>, Error> {
    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let rec = to_upd
        .save_changes::<User>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn user_get(id: PathParam<i64>, depot: &Depot) -> Result<Json<User>, Error> {
    use crate::schema::users;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let rec = users::table
        .find(id.clone())
        .get_result::<User>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn user_del(id: PathParam<i64>, depot: &Depot) -> Result<(), Error> {
    use crate::schema::users;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let _ = diesel::delete(users::table.filter(users::id.eq(id.clone())))
        .execute(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(())
}

#[endpoint]
pub async fn user_lis(
    cur: QueryParam<u64, true>,
    num: QueryParam<u64, true>,
    flt: JsonBody<UserToLis>,
    depot: &Depot,
) -> Result<Json<LisRes<User>>, Error> {
    use crate::schema::users;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    conn.transaction::<_, diesel::result::Error, _>(|conn| {
        let flt_inner = flt.clone();
        let get_query = || {
            let mut query = users::table.into_boxed();
            if let Some(username) = flt_inner.clone().username {
                query = query.filter(users::username.like(format!("%{}%", username)));
            }
            query
        };
        let recs = get_query()
            .limit(num.clone() as i64)
            .offset((cur.clone() * num.clone()) as i64)
            .load::<User>(conn)?;
        let total = get_query().count().get_result::<i64>(conn)?;
        Ok(Json(LisRes { data: recs, total }))
    })
    .map_err(|e| Error::Anyhow(e.into()))
}

#[derive(Deserialize, ToSchema)]
pub struct SpaceToAddPartial {
    pub name: String,
    pub desp: Option<String>,
    pub booking: bool,
}

#[endpoint]
pub async fn space_add(
    to_add_partial: JsonBody<SpaceToAddPartial>,
    depot: &Depot,
) -> Result<Json<Space>, Error> {
    use crate::schema::spaces;

    let owner_id = obtain_user_id(depot)?;
    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let to_add = SpaceToAdd {
        owner_id,
        name: to_add_partial.name.to_owned(),
        desp: to_add_partial.desp.to_owned(),
        booking: to_add_partial.booking,
        state: serde_json::to_value(SpaceState::default())?,
    };
    let rec = diesel::insert_into(spaces::table)
        .values(Some(to_add))
        .get_result::<Space>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn space_upd(to_upd: FormBody<SpaceToUpd>, depot: &Depot) -> Result<Json<Space>, Error> {
    use crate::schema::spaces;

    let user_id = obtain_user_id(depot)?;
    let mut redis = obtain_redis(depot)?;
    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let space = conn
        .transaction::<_, anyhow::Error, _>(|conn| {
            let space_old = spaces::table.find(to_upd.id).get_result::<Space>(conn)?;
            if user_id == space_old.owner_id {
                let space_new = to_upd.save_changes::<Space>(conn)?;
                Ok(space_new)
            } else {
                bail!("space can only be updated by its owner.")
            }
        })
        .map_err(|e| Error::Anyhow(e.into()))?;
    let _ = redis
        .connection_manager
        .json_set::<_, _, _, ()>(space_key_basic(to_upd.id), "$", &space)
        .await
        .map_err(|e| anyhow!(e))?;
    Ok(Json(space))
}

#[endpoint]
pub async fn space_get(id: PathParam<i64>, depot: &Depot) -> Result<Json<Space>, Error> {
    use crate::schema::spaces;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let rec = spaces::table
        .find(id.clone())
        .get_result::<Space>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn space_del(id: PathParam<i64>, depot: &Depot) -> Result<(), Error> {
    use crate::schema::spaces;

    let user_id = obtain_user_id(depot)?;
    let redis = obtain_redis(depot)?;
    if load_space_is_online(redis, id.clone()).await? {
        return Err(Error::Anyhow(anyhow!("不能删除正在进行的课程")));
    }
    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let _ = diesel::delete(
        spaces::table.filter(spaces::id.eq(id.clone()).and(spaces::owner_id.eq(user_id))),
    )
    .execute(&mut conn)
    .map_err(|e| anyhow!(e))?;
    Ok(())
}

#[derive(Serialize, ToSchema)]
pub struct SpaceBasicPlus {
    pub id: i64,
    pub owner_id: i64,
    pub owner_name: String,
    pub name: String,
    pub desp: Option<String>,
    pub booking: bool,
    pub booking_status: Option<bool>,
}

#[endpoint]
pub async fn space_lis(
    cur: QueryParam<u64, true>,
    num: QueryParam<u64, true>,
    flt: JsonBody<SpaceToLis>,
    depot: &Depot,
) -> Result<Json<LisRes<SpaceBasicPlus>>, Error> {
    use crate::schema::bookings;
    use crate::schema::spaces;
    use crate::schema::users;

    let user_id = obtain_user_id(depot)?;
    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    conn.transaction::<_, diesel::result::Error, _>(|conn| {
        let flt_inner = flt.clone();
        let get_query = || {
            let mut query = spaces::table.into_boxed();
            if let Some(owner_id) = flt_inner.clone().owner_id {
                query = query.filter(spaces::owner_id.eq(owner_id));
            }
            if let Some(name) = flt_inner.clone().name {
                query = query.filter(spaces::name.like(format!("%{}%", name)));
            }
            if let Some(desp) = flt_inner.clone().desp {
                query = query.filter(spaces::desp.like(format!("%{}%", desp)));
            }
            if let Some(booking) = flt_inner.clone().booking {
                query = query.filter(spaces::booking.eq(booking));
            }
            query
        };
        let recs = get_query()
            .inner_join(users::table)
            .left_join(
                bookings::table.on(bookings::space_id
                    .eq(spaces::id)
                    .and(bookings::user_id.eq(user_id))),
            )
            .select((
                Space::as_select(),
                User::as_select(),
                bookings::status.nullable(),
            ))
            .limit(num.clone() as i64)
            .offset((cur.clone() * num.clone()) as i64)
            .load::<(Space, User, Option<bool>)>(conn)?
            .iter()
            .map(|(space, user, booking_status)| SpaceBasicPlus {
                id: space.id,
                owner_id: space.owner_id,
                owner_name: user.username.clone(),
                name: space.name.clone(),
                desp: space.desp.clone(),
                booking: space.booking,
                booking_status: booking_status.clone(),
            })
            .collect();
        let total = get_query().count().get_result::<i64>(conn)?;
        Ok(Json(LisRes { data: recs, total }))
    })
    .map_err(|e| Error::Anyhow(e.into()))
}

#[endpoint]
pub async fn space_enter(space_id: PathParam<i64>, depot: &Depot) -> Result<Json<()>, Error> {
    let user_id = obtain_user_id(depot)?;
    let db = obtain_db(depot)?;
    let redis = obtain_redis(depot)?;
    let mqtt_client = obtain_mqtt_client(depot)?;
    let space = load_space_basic(db.clone(), redis.clone(), space_id.clone()).await?;
    if user_id == space.owner_id {
        let _ = sync_space_active(redis.clone(), space_id.clone()).await?;
        let _ = sync_space_user_perm(
            redis.clone(),
            mqtt_client,
            space.clone(),
            user_id,
            PermRole::Teacher,
        )
        .await?;
        let _ = load_space_state(db.clone(), redis.clone(), space_id.clone()).await?;
    } else {
        if !load_space_is_online(redis.clone(), space_id.clone()).await? {
            return Err(Error::Anyhow(anyhow!("等待开课")));
        }
        let mut allow_user = true;
        if space.booking {
            allow_user = user_booking_granted(db.clone(), space_id.clone(), user_id).await?;
        }
        if !allow_user {
            return Err(Error::Anyhow(anyhow!("需要预约")));
        }
        let _ = sync_space_user_perm(
            redis.clone(),
            mqtt_client,
            space,
            user_id,
            PermRole::Watcher,
        )
        .await?;
    }
    let _ = load_space_user_basic(db, redis.clone(), space_id.clone(), user_id).await?;
    let _ = sync_space_user_ping(redis.clone(), space_id.clone(), user_id).await?;
    let _ = sync_space_online(redis, space_id.clone()).await?;
    Ok(Json(()))
}

#[endpoint]
pub async fn space_leave(space_id: PathParam<i64>, depot: &Depot) -> Result<Json<()>, Error> {
    let user_id = obtain_user_id(depot)?;
    let db = obtain_db(depot)?;
    let redis = obtain_redis(depot)?;
    let mqtt_client = obtain_mqtt_client(depot)?;
    let space = load_space_basic(db.clone(), redis.clone(), space_id.clone()).await?;
    info!("{} - {}", user_id, space.owner_id);
    if space.owner_id != user_id {
        let _ = drop_space_user_perm(redis.clone(), mqtt_client, space, user_id).await?;
        let _ = drop_space_user_basic(redis.clone(), space_id.clone(), user_id).await?;
        let _ = drop_space_user_ping(redis, space_id.clone(), user_id).await?;
    }
    Ok(Json(()))
}

#[endpoint]
pub async fn space_state(
    space_id: PathParam<i64>,
    depot: &Depot,
) -> Result<Json<SpaceState>, Error> {
    let user_id = obtain_user_id(depot)?;
    let db = obtain_db(depot)?;
    let redis = obtain_redis(depot)?;
    let user_ids = list_space_perms(redis.clone(), space_id.clone())
        .await?
        .iter()
        .map(|(uid, _)| uid.clone())
        .collect::<Vec<i64>>();
    if user_ids.contains(&user_id) {
        let state = load_space_state(db.clone(), redis, space_id.clone()).await?;
        Ok(Json(state))
    } else {
        Err(Error::Anyhow(anyhow!("access forbidden")))
    }
}

#[endpoint]
pub async fn space_state_patch(
    space_id: PathParam<i64>,
    payload: JsonBody<PayloadPatch>,
    depot: &Depot,
) -> Result<Json<()>, Error> {
    let db = obtain_db(depot)?;
    let redis = obtain_redis(depot)?;
    let _ = patch_state(space_id.clone(), db.clone(), redis, &payload.into_inner())
        .await
        .map_err(|e| Error::Anyhow(e.into()))?;
    Ok(Json(()))
}

async fn user_booking_granted(db: Db, space_id: i64, user_id: i64) -> anyhow::Result<bool> {
    use crate::schema::bookings;

    let mut conn = db.connect()?;
    let count = bookings::table
        .filter(
            bookings::space_id
                .eq(space_id)
                .and(bookings::user_id.eq(user_id))
                .and(bookings::status.eq(true)),
        )
        .count()
        .get_result::<i64>(&mut conn)?;
    Ok(count == 1)
}

#[endpoint]
pub async fn booking_add(
    to_add: JsonBody<BookingToAdd>,
    depot: &Depot,
) -> Result<Json<Booking>, Error> {
    use crate::schema::bookings;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let rec = diesel::insert_into(bookings::table)
        .values(Some(to_add.clone()))
        .get_result::<Booking>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn booking_upd(
    to_upd: FormBody<BookingToUpd>,
    depot: &Depot,
) -> Result<Json<Booking>, Error> {
    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let rec = to_upd
        .save_changes::<Booking>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn booking_get(
    space_id: PathParam<i64>,
    user_id: PathParam<i64>,
    depot: &Depot,
) -> Result<Json<Booking>, Error> {
    use crate::schema::bookings;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let rec = bookings::table
        .find((space_id.clone(), user_id.clone()))
        .get_result::<Booking>(&mut conn)
        .map_err(|e| anyhow!(e))?;
    Ok(Json(rec))
}

#[endpoint]
pub async fn booking_del(
    space_id: PathParam<i64>,
    user_id: PathParam<i64>,
    depot: &Depot,
) -> Result<(), Error> {
    use crate::schema::bookings;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    let _ = diesel::delete(
        bookings::table.filter(
            bookings::space_id
                .eq(space_id.clone())
                .and(bookings::user_id.eq(user_id.clone())),
        ),
    )
    .execute(&mut conn)
    .map_err(|e| anyhow!(e))?;
    Ok(())
}

#[endpoint]
pub async fn booking_lis(
    cur: QueryParam<u64, true>,
    num: QueryParam<u64, true>,
    flt: JsonBody<BookingToLis>,
    depot: &Depot,
) -> Result<Json<LisRes<Booking>>, Error> {
    use crate::schema::bookings;

    let mut conn = obtain_db(depot)?.connect().map_err(|e| anyhow!(e))?;
    conn.transaction::<_, diesel::result::Error, _>(|conn| {
        let flt_inner = flt.clone();
        let get_query = || {
            let mut query = bookings::table.into_boxed();
            if let Some(space_id) = flt_inner.clone().space_id {
                query = query.filter(bookings::space_id.eq(space_id));
            }
            if let Some(user_id) = flt_inner.clone().user_id {
                query = query.filter(bookings::user_id.eq(user_id));
            }
            if let Some(status) = flt_inner.clone().status {
                query = query.filter(bookings::status.eq(status));
            }
            query
        };
        let recs = get_query()
            .limit(num.clone() as i64)
            .offset((cur.clone() * num.clone()) as i64)
            .load::<Booking>(conn)?;
        let total = get_query().count().get_result::<i64>(conn)?;
        Ok(Json(LisRes { data: recs, total }))
    })
    .map_err(|e| Error::Anyhow(e.into()))
}
